package log.job;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class StockDataJob {
    public static void main(String[] args) {
//  todo      1创建流式环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()  // 流式处理模式
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);

        try{
//  todo          2注册kafka表
            registerKafkaTables(tabEnv);
//  todo          3 连表

            Table resultTable = executeJoinQuery(tabEnv);
//  todo          4 输出结果

//            resultTable.execute().print();

//  todo          5 输出表
            sinkToKafka(tabEnv, resultTable);
//  todo          6 保存数据
            saveTosink(tabEnv,resultTable);


        }catch (Exception e){
            e.printStackTrace();
        }




    }

    private static void saveTosink(TableEnvironment tabEnv, Table resultTable) {
        tabEnv.createTemporaryView("stock", resultTable);
//        插入
        tabEnv.executeSql(
                "INSERT INTO dwd_stock_join\n" +
                        "SELECT\n" +
                        "stock_code ,\n" +
                        "DATE_FORMAT(data_event_time, 'yyyy-MM-dd HH:mm:ss.SSS') AS data_event_time ,\n" +
                        "stock_name ,\n" +
                        "trade_calendar_date ,\n" +
                        "trade_year ,\n" +
                        "trade_month ,\n" +
                        "trade_day ,\n" +
                        "trade_weekday ,\n" +
                        "is_trade_workday ,\n" +
                        "is_trade_holiday ,\n" +
                        "realtime_trade_vol ,\n" +
                        "realtime_trade_amt ,\n" +
                        "realtime_pre_close ,\n" +
                        "realtime_open_price ,\n" +
                        "realtime_high_price ,\n" +
                        "realtime_low_price ,\n" +
                        "realtime_current_price ,\n" +
                        "trade_phase ,\n" +
                        "daily_close_price ,\n" +
                        "daily_total_vol ,\n" +
                        "daily_total_amt ,\n" +
                        "daily_total_market_cap ,\n" +
                        "daily_circulating_cap ,\n" +
                        "daily_kline_type ,\n" +
                        "daily_avg_price ,\n" +
                        "data_update_ts\n" +
                        "FROM stock"
        );

    }

    private static void sinkToKafka(TableEnvironment tabEnv, Table resultTable) {
        // 创建输出表
        String sinkDdl = "CREATE TABLE dwd_stock_join (\n" +
                "    stock_code String,\n" +
                "    data_event_time String,\n" +
                "    stock_name String,\n" +
                "    trade_calendar_date String,\n" +
                "    trade_year String,\n" +
                "    trade_month String,\n" +
                "    trade_day String,\n" +
                "    trade_weekday Bigint,\n" +
                "    is_trade_workday String,\n" +
                "    is_trade_holiday String,\n" +
                "    realtime_trade_vol Bigint,\n" +
                "    realtime_trade_amt Double,\n" +
                "    realtime_pre_close Double,\n" +
                "    realtime_open_price Double,\n" +
                "    realtime_high_price Double,\n" +
                "    realtime_low_price Double,\n" +
                "    realtime_current_price Double,\n" +
                "    trade_phase String,\n" +
                "    daily_close_price Double,\n" +
                "    daily_total_vol Double,\n" +
                "    daily_total_amt Double,\n" +
                "    daily_total_market_cap Bigint,\n" +
                "    daily_circulating_cap Bigint,\n" +
                "    daily_kline_type String,\n" +
                "    daily_avg_price Double,\n" +
                "    data_update_ts Bigint\n" +
                ") WITH (\n" +
                "  'connector'='doris',\n" +
                "    'fenodes'='node102:8030',\n" +
                "    'table.identifier'='gd1_gp.dwd_stock_join',\n" +
                "    'username'='root',\n" +
                "    'password'='123456',\n" +
                "    'sink.label-prefix'='doris_label'\n" +
                ")";
        tabEnv.executeSql(sinkDdl);



    }

    //连表
    private static Table executeJoinQuery(TableEnvironment tabEnv) {
//        关联sql
        String sql =
                "SELECT \n" +
                        "  -- 1. 股票基础标识\n" +
                        "  t.securityID AS stock_code,\n" +
                        "  t.symbol AS stock_name,\n" +
                        "  -- 2. 时间维度\n" +
                        "  d.date_dd AS trade_calendar_date,\n" +
                        "  d.year_yy AS trade_year,\n" +
                        "  d.month_mm AS trade_month,\n" +
                        "  d.day_dd AS trade_day,\n" +
                        "  d.day_of_week AS trade_weekday,\n" +
                        "  d.is_workday AS is_trade_workday,\n" +
                        "  d.is_holiday AS is_trade_holiday,\n" +
                        "  -- 3. 实时交易数据\n" +
                        "  SUM(t.tradeVolume) AS realtime_trade_vol,\n" +
                        "  SUM(t.totalValueTraded) AS realtime_trade_amt,\n" +
                        "  MAX(t.preClosePx) AS realtime_pre_close,\n" +
                        "  MAX(t.openPrice) AS realtime_open_price,\n" +
                        "  MAX(t.highPrice) AS realtime_high_price,\n" +
                        "  MIN(t.lowPrice) AS realtime_low_price,\n" +
                        "  -- 窗口函数：取窗口内最新价格\n" +
                        "  LAST_VALUE(t.tradePrice) OVER (\n" +
                        "    PARTITION BY t.securityID, t.window_start, t.window_end \n" +
                        "    ORDER BY t.update_event_time \n" +
                        "    ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING\n" +
                        "  ) AS realtime_current_price,\n" +
                        "  MAX(t.tradingPhaseCode) AS trade_phase,\n" +
                        "  -- 4. 每日收盘数据\n" +
                        "  c.cur_close_price AS daily_close_price,\n" +
                        "  c.cur_trade_vol AS daily_total_vol,\n" +
                        "  c.cur_trade_amt AS daily_total_amt,\n" +
                        "  c.tot_cap AS daily_total_market_cap,\n" +
                        "  c.nego_cap AS daily_circulating_cap,\n" +
                        "  -- 5. K线数据\n" +
                        "  k.avgprice AS daily_avg_price,\n" +
                        "  -- 6. 时间戳与窗口信息\n" +
                        "  MAX(t.update_timestamp) AS data_update_ts,\n" +
                        "  t.window_start AS window_start_time,\n" +
                        "  t.window_end AS window_end_time,\n" +
                        "  DATE_FORMAT(t.window_start, 'yyyy-MM-dd HH:mm:ss') AS window_start_str,\n" +
                        "  DATE_FORMAT(t.window_end, 'yyyy-MM-dd HH:mm:ss') AS window_end_str\n" +
                        "-- HOP窗口函数：无多余逗号，参数格式正确\n" +
                        "FROM HOP(stock_transaction_records, update_event_time, INTERVAL '1' SECOND, INTERVAL '5' SECOND) AS t\n" +
                        "-- 关联1：股票信息表\n" +
                        "LEFT JOIN stock_info i \n" +
                        "  ON t.securityID = i.stock_code\n" +
                        "  AND i.stock_code IS NOT NULL\n" +
                        "-- 关联2：日期维度表\n" +
                        "LEFT JOIN transaction_date_dim d \n" +
                        "  ON DATE_FORMAT(t.update_event_time, 'yyyyMMdd') = d.date_dd\n" +
                        "  AND d.is_workday = 1\n" +
                        "-- 关联3：每日收盘表\n" +
                        "LEFT JOIN day_closing_stocks c \n" +
                        "  ON t.securityID = c.sec_code\n" +
                        "  AND DATE_FORMAT(t.update_event_time, 'yyyyMMdd') = DATE_FORMAT(c.trade_date, 'yyyyMMdd')\n" +
                        "  AND c.sec_code IS NOT NULL\n" +
                        "-- 关联4：个股K线表\n" +
                        "LEFT JOIN stock_K_line k \n" +
                        "  ON t.securityID = k.sec_code\n" +
                        "  AND DATE_FORMAT(t.update_event_time, 'yyyyMMdd') = DATE_FORMAT(k.trade_date, 'yyyyMMdd')\n" +
                        "  AND k.k_line_type = 'DAY'\n" +
                        "  AND k.sec_code IS NOT NULL\n" +
                        "-- 关键修正：GROUP BY无多余逗号（第40行附近无尾随逗号）\n" +
                        "GROUP BY \n" +
                        "  t.securityID, \n" +
                        "  t.symbol,\n" +
                        "  d.date_dd, \n" +
                        "  d.year_yy, \n" +
                        "  d.month_mm, \n" +
                        "  d.day_dd, \n" +
                        "  d.day_of_week, \n" +
                        "  d.is_workday, \n" +
                        "  d.is_holiday,\n" +
                        "  c.cur_close_price, \n" +
                        "  c.cur_trade_vol, \n" +
                        "  c.cur_trade_amt,\n" +
                        "  c.tot_cap, \n" +
                        "  c.nego_cap,\n" +
                        "  k.avgprice,\n" +
                        "  t.window_start, \n" +
                        "  t.window_end  -- 此处无逗号！这是第40行附近的核心修正点\n" +
                        "-- 过滤无交易数据的窗口\n" +
                        "HAVING SUM(t.tradeVolume) > 0";

        Table table = tabEnv.sqlQuery(sql);
        return table;

//        String join="SELECT \n" +
//                "    *\n" +
//                "FROM transaction_date_dim";
//        Table table = tabEnv.sqlQuery(join);
//        return table;

    }

    private static void registerKafkaTables(TableEnvironment tabEnv) {
//  todo      股票基本信息
        String stockinfo = "CREATE TABLE stock_info (\n" +
                "      stock_code String,\n" +
                "      stock_name String,\n" +
                "      market String,\n" +
                "      sector String,\n" +
                "      industry String,\n" +
                "      listing_date String,\n" +
                "      total_shares Bigint,\n" +
                "      circulating_shares Bigint,\n" +
                "      is_st String,\n" +
                "      is_suspended String,\n" +
                "      update_time DATE,\n" +
                "      update_timestamp Bigint,\n" +
                "      update_event_time AS TO_TIMESTAMP_LTZ(update_timestamp, 3),\n" +
                "  WATERMARK FOR update_event_time AS update_event_time - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'stock_info',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "  'properties.group.id' = 'group',\n" +
                "  'format' = 'json',\n" +
                "    'properties.auto.offset.reset' = 'earliest',\n" +
                "  'json.fail-on-missing-field' = 'false',\n" +
                "  'json.ignore-parse-errors' = 'true'\n" +
                ");\n";
        tabEnv.executeSql(stockinfo);
//  todo       股票交易记录数据
        String stocktransactionrecords ="CREATE TABLE stock_transaction_records (\n" +
                        "   securityID String,\n" +
                        "   symbol String,\n" +
                        "   tradeVolume Bigint,\n" +
                        "   totalValueTraded Double,\n" +
                        "   preClosePx Double,\n" +
                        "   openPrice Double,\n" +
                        "   highPrice Double,\n" +
                        "   lowPrice Double,\n" +
                        "   tradePrice Double,\n" +
                        "   closePx String,\n" +
                        "   tradingPhaseCode String,\n" +
                        "   update_timestamp Bigint,\n" +
                        " update_event_time AS TO_TIMESTAMP_LTZ(update_timestamp, 3),\n" +
                        "  WATERMARK FOR update_event_time AS update_event_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'stock_transaction_records',\n" +
                        "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.auto.offset.reset' = 'earliest',\n" +
                        "  'properties.group.id' = 'group',\n" +
                        "  'format' = 'json'\n" +
                        ")";
        tabEnv.executeSql(stocktransactionrecords);
//  todo      交易日期维度表
        String dateDimDdl
                = "CREATE TABLE transaction_date_dim (\n" +
                "  stock_code String,\n" +
                "  stock_name String,\n" +
                "  date_dd String,\n" +
                "  year_yy  String,\n" +
                "  month_mm String,\n" +
                "  day_dd String,\n" +
                "  day_of_week Bigint,\n" +
                "  week_of_year Bigint,\n" +
                "  is_workday String,\n" +
                "  is_holiday String,\n" +
                "  update_timestamp Bigint,\n" +
                " update_event_time AS TO_TIMESTAMP_LTZ(update_timestamp, 3),\n" +
                "  WATERMARK FOR update_event_time AS update_event_time - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'transaction_date_dim',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "    'properties.auto.offset.reset' = 'earliest',\n" +
                "  'properties.group.id' = 'group',\n" +
                "  'format' = 'json'\n" +
                ")";
        tabEnv.executeSql(dateDimDdl);
//  todo      股票每日收盘明细表( day_closing_stocks)
        String dayclosingstocks="CREATE TABLE day_closing_stocks (\n" +
                        "    trade_date String,\n" +
                        "    sec_code String,\n" +
                        "    sec_abbr String,\n" +
                        "    last_close_price Double,\n" +
                        "    cur_open_price Double,\n" +
                        "    cur_high_price Double,\n" +
                        "    cur_low_price Double,\n" +
                        "    cur_close_price Double,\n" +
                        "    cur_trade_vol Double,\n" +
                        "    cur_trade_amt Double,\n" +
                        "    tot_cap Bigint,\n" +
                        "    nego_cap Bigint,\n" +
                        "    update_timestamp Bigint,\n" +
                        " update_event_time AS TO_TIMESTAMP_LTZ(update_timestamp, 3),\n" +
                        "  WATERMARK FOR update_event_time AS update_event_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "      'connector' = 'kafka',\n" +
                        "      'topic' = 'day_closing_stocks',\n" +
                        "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.auto.offset.reset' = 'earliest',\n" +
                        "  'properties.group.id' = 'group',\n" +
                        "      'format' = 'json'\n" +
                        "      )";
        tabEnv.executeSql(dayclosingstocks);

//  todo       个股K线表 (stock_K_line)
        String kLineDdl ="CREATE TABLE stock_K_line (\n" +
                        "    update_time String,\n" +
                        "    trade_date DATE,\n" +
                        "    sec_code String,\n" +
                        "    sec_name String,\n" +
                        "    k_line_type String,\n" +
                        "    pre_close_price Double,\n" +
                        "    open_price Double,\n" +
                        "    high_price Double,\n" +
                        "    low_price Double,\n" +
                        "    close_price Double,\n" +
                        "    avgprice Double,\n" +
                        "    trade_vol Double,\n" +
                        "    trade_amt Double,\n" +
                        "    update_timestamp Bigint,\n" +
                        " update_event_time AS TO_TIMESTAMP_LTZ(update_timestamp, 3),\n" +
                        "  WATERMARK FOR update_event_time AS update_event_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'stock_K_line',\n" +
                        "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.auto.offset.reset' = 'earliest',\n" +
                        "  'properties.group.id' = 'group',\n" +
                        "  'format' = 'json'\n" +
                        ")";
        tabEnv.executeSql(kLineDdl);
    }
}
